package org.apache.rocketmq.client.java.impl;


import apache.rocketmq.v2.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.stub.StreamObserver;
import org.apache.rocketmq.client.java.clientapis.ClientException;
import org.apache.rocketmq.client.java.impl.producer.ClientSessionHandler;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.concurrent.TimeUnit;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/10
 * @方法描述：客户端会话管理器，这个会话管理器就负责维护管理客户端和服务端建立的连接，也就是grpc的双向流
 * 并且客户端与服务端的TelemetryCommand遥测请求都是通过这个会话管理器发送和接收的
 */
public class ClientSessionImpl implements StreamObserver<TelemetryCommand> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ClientSessionImpl.class);

    //重建连接的间隔时间，默认为1秒
    private static final Duration REQUEST_OBSERVER_RENEW_BACKOFF_DELAY = Duration.ofSeconds(1);

    //会话处理器，这里的这个会话处理器其实就是ClientImpl对象
    private final ClientSessionHandler sessionHandler;

    //生产者客户端建立会话连接的服务端的网络地址
    private final Endpoints endpoints;

    //grpc的双向流中的请求观察者对象，生产者和指定的客户端建立网络连接后，这个请求观察者对象也就创建了
    //是用这个请求观察者对象可以向服务端发送请求
    //而当前类实现了StreamObserver接口，这也就意味着当前类的对象就是一个观察者对象，只不过这个观察者对象会接收从服务端发送过来的TelemetryCommand遥测请求
    //当前类对象和下面这个成员变量构成了Grpc双向流
    private volatile StreamObserver<TelemetryCommand> requestObserver;

    //构造方法
    protected ClientSessionImpl(ClientSessionHandler sessionHandler, Endpoints endpoints) throws ClientException {
        this.sessionHandler = sessionHandler;
        this.endpoints = endpoints;
        //这里可以看到，在创建ClientSessionImpl对象时，会调用ClientImpl对象的telemetry方法
        //这个方法一旦被调用，生产者客户端就会和服务端建立网络连接，所以这个方法会返回一个请求观察者对象
        this.requestObserver = sessionHandler.telemetry(endpoints, this);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：重建客户端和服务端链接的方法
     */
    private void renewRequestObserver() {
        try {
            //首先判断一下客户端要建立网络连接的服务端的网络地址是否已被废弃
            if (sessionHandler.isEndpointsDeprecated(endpoints)) {
                //如果已被废弃则记录日志信息，然后退出当前方法
                LOGGER.info("Endpoints is deprecated, no longer to renew requestObserver, endpoints={}", endpoints);
                return;
            }
            //没有被废弃则执行ClientImpl对象的telemetry方法，这个方法会重新建立客户端与服务端的网络连接
            //这样一来，就可以得到新的请求观察者对象了
            this.requestObserver = sessionHandler.telemetry(endpoints, this);
        } catch (Throwable t) {
            //重连失败则记录错误日志
            LOGGER.error("Failed to renew requestObserver, attempt to renew later, endpoints={}, delay={}", endpoints,
                    REQUEST_OBSERVER_RENEW_BACKOFF_DELAY, t);
            //在指定时间之后重新执行当前方法，再次尝试重连服务端
            sessionHandler.getScheduler().schedule(this::renewRequestObserver,
                    REQUEST_OBSERVER_RENEW_BACKOFF_DELAY.toNanos(), TimeUnit.NANOSECONDS);
            return;
        }
        //连接重建成功后，客户端要把自己的配置信息同步给服务端
        syncSettings();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：安全同步客户端配置信息到服务端的方法
     */
    protected ListenableFuture<Void> syncSettingsSafely() {
        try {
            //在这里执行了同步客户端配置信息到服务端的操作
            this.syncSettings();
            //等待客户端同步配置信息操作完成，这里会返回一个表示同步状态的Future对象
            //返回的实际上就是ClientSettings对象的arrivedFuture成员变量
            //而在接收到服务端回复的响应后，这个arrivedFuture对象就会被设置完成状态
            //如果这个arrivedFuture对象一直没有被设置完成状态，那么调用了arrivedFuture.get()方法的线程就会一直阻塞
            //这一点我会在文章中详细讲解
            return sessionHandler.awaitSettingSynchronized();
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：同步客户端设置信息到服务端的方法
     */
    private void syncSettings() {
        //这里得到了封装了客户端所有设置信息的TelemetryCommand对象
        final TelemetryCommand settings = sessionHandler.settingsCommand();
        //是用请求观察者对象把这个TelemetryCommand对象发送到服务端
        fireWrite(settings);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：释放请求观察者对象的方法
     */
    public void release() {
        if (null == requestObserver) {
            return;
        }
        try {
            requestObserver.onCompleted();
        } catch (Throwable ignore) {
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：发送遥测请求给服务端的方法
     */
    void fireWrite(TelemetryCommand command) {
        if (null == requestObserver) {
            LOGGER.error("Request observer does not exist, ignore current command, endpoints={}, command={}",
                    endpoints, command);
            return;
        }
        requestObserver.onNext(command);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：接收并处理来自服务端的遥测请求的方法
     */
    @Override
    public void onNext(TelemetryCommand command) {
        //得到客户端的唯一Id
        final String clientId = sessionHandler.clientId();
        try {
            //根据遥测请求的类型判断要执行的操作
            switch (command.getCommandCase()) {
                case SETTINGS: {
                    //执行到这里意味着服务端发送过来的是设置命令，也就是说服务端向客户端发送了一个TelemetryCommand请求
                    //这个请求中包含的是服务端向客户端发送的客户端设置信息，当客户端接收到这些信息后，要把从这些设置信息同步到自己本地
                    final Settings settings = command.getSettings();
                    LOGGER.info("Receive settings from remote, endpoints={}, clientId={}", endpoints, clientId);
                    //将服务端发送过来的设置信息同步到自己本地
                    sessionHandler.onSettingsCommand(endpoints, settings);
                    break;
                }
                //以下遥测请求类型在当前版本代码中用不上，所以我就先不添加注释了，大家感兴趣的话可以自己看看，内容都很简单
                case RECOVER_ORPHANED_TRANSACTION_COMMAND: {
                    final RecoverOrphanedTransactionCommand recoverOrphanedTransactionCommand =
                            command.getRecoverOrphanedTransactionCommand();
                    LOGGER.info("Receive orphaned transaction recovery command from remote, endpoints={}, "
                            + "clientId={}", endpoints, clientId);
                    sessionHandler.onRecoverOrphanedTransactionCommand(endpoints, recoverOrphanedTransactionCommand);
                    break;
                }
                case VERIFY_MESSAGE_COMMAND: {
                    final VerifyMessageCommand verifyMessageCommand = command.getVerifyMessageCommand();
                    LOGGER.info("Receive message verification command from remote, endpoints={}, clientId={}",
                            endpoints, clientId);
                    sessionHandler.onVerifyMessageCommand(endpoints, verifyMessageCommand);
                    break;
                }
                case PRINT_THREAD_STACK_TRACE_COMMAND: {
                    final PrintThreadStackTraceCommand printThreadStackTraceCommand =
                            command.getPrintThreadStackTraceCommand();
                    LOGGER.info("Receive thread stack print command from remote, endpoints={}, clientId={}",
                            endpoints, clientId);
                    sessionHandler.onPrintThreadStackTraceCommand(endpoints, printThreadStackTraceCommand);
                    break;
                }
                default:
                    //这里就是未知的遥测请求类型，记录日志即可
                    LOGGER.warn("Receive unrecognized command from remote, endpoints={}, command={}, clientId={}",
                            endpoints, command, clientId);
            }
        } catch (Throwable t) {
            LOGGER.error("[Bug] unexpected exception raised while receiving command from remote, command={}, "
                    + "clientId={}", command, clientId, t);
        }
    }


    @Override
    public void onError(Throwable throwable) {
        LOGGER.error("Exception raised from stream response observer, clientId={}, endpoints={}",
                sessionHandler.clientId(), endpoints, throwable);
        release();
        if (!sessionHandler.isRunning()) {
            return;
        }
        sessionHandler.getScheduler().schedule(this::renewRequestObserver, 3, TimeUnit.SECONDS);
    }


    @Override
    public void onCompleted() {
        release();
        if (!sessionHandler.isRunning()) {
            return;
        }
        sessionHandler.getScheduler().schedule(this::renewRequestObserver, 3, TimeUnit.SECONDS);
    }
}
